package top.toknow.storm.demo.Bolt;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * @desc 描述
 * create on 2018/10/30 by watson
 */
public class MyBolt extends BaseBasicBolt {

    private OutputCollector collector;


    private Integer index = 0;

    /**
     * 在Bolt启动前执行，提供Bolt启动环境配置的入口。
     * 参数基本和Sqout一样。
     * 一般对于不可序列化的对象进行实例化。
     * 这里的我们就简单的打印下
     *
     * @param stormConf
     * @param context
     */
    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        System.out.println("prepare:"+ stormConf.get("test01"));
    }

    /**
     * execute()方法是Bolt实现的核心。
     * 也就是执行方法，每次Bolt从流接收一个订阅的tuple，都会调用这个方法。
     * 从tuple中获取消息可以使用 tuple.getString()和tuple.getStringByField();这两个方法。个人推荐第二种，可以通过field来指定接收的消息。
     * 注:如果继承的是IRichBolt，则需要手动ack。这里就不用了,BaseRichBolt会自动帮我们应答。
     *
     * @param tuple
     * @param basicOutputCollector
     */
    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
//      String msg=tuple.getString(0);
        String msg=tuple.getStringByField("test01");
        //这里我们就不做消息的处理，只打印
        System.out.println("Bolt第"+index+"接受的消息:"+msg);
        index++;
        /**
         *
         * 没次调用处理一个输入的tuple，所有的tuple都必须在一定时间内应答。
         * 可以是ack或者fail。否则，spout就会重发tuple。
         */
//      collector.ack(tuple);
    }

    /**
     * 和Spout的一样。
     * 因为到了这里就不再输出了，所以就什么都没写。
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    /**
     * cleanup是IBolt接口中定义,用于释放bolt占用的资源。
     * Storm在终止一个bolt之前会调用这个方法。
     * 因为这里没有什么资源需要释放，所以就简单的打印一句就行了。
     */
    @Override
    public void cleanup() {
        System.out.println("资源释放");
    }


}
